package com.tianji.aigc.service.Impl;


import com.tianji.aigc.config.SpringAiConfig;
import com.tianji.aigc.domain.dto.ChatDTO;
import com.tianji.aigc.domain.vo.ChatEventVO;
import com.tianji.aigc.enums.ChatEventTypeEnum;
import com.tianji.aigc.service.ChatService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.chat.model.ChatResponse;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import java.util.function.Predicate;

@Slf4j
@Service
@RequiredArgsConstructor
public class ChatServiceImpl implements ChatService {

    @Autowired
    private ChatClient chatClient;

    // 存储大模型的生成状态，这里采用ConcurrentHashMap是确保线程安全
    // 目前的版本暂时用Map实现，如果考虑分布式环境的话，可以考虑用redis来实现
    //key放sessionId，value：当前是否继续输出的标识
    //如果是分步式环境的话，
    //volatile java的关键字
    private static final Map<String, Boolean> FLAG_MAP = new ConcurrentHashMap<>();


    /**
     * 聊天
     *
     * @param question  问题
     * @param sessionId 会话id
     * @return 回答内容
     * 响应式编程 流式返回数据 异步处理
     */
    @Override
    public Flux<ChatEventVO> chat(ChatDTO chatDTO) {
        String sessionId = chatDTO.getSessionId();
        return chatClient.prompt()
                .system(SpringAiConfig.SYSTEM_PROMPT)
                .user(chatDTO.getQuestion())
                //调用stream()方法，会返回一个Flux对象，这个对象会流式返回数据
                .stream()
                //创建一个ChatEventVO对象，将数据封装进去
                //获取聊天响应的数据
                .chatResponse()
                //流第一次开始执行的时候,大模型开始返回第一个token时候
                .doFirst(new Runnable() {
                    @Override
                    public void run() {
                        log.info("doFirst开始,第一次执行:{}", sessionId);
                        FLAG_MAP.put(sessionId, true);
                    }
                })
                //流完成的时候执行
                .doOnComplete(new Runnable() {
                    @Override
                    public void run() {
                        log.info("doOnComplete结束的时候开始,完成:{}", sessionId);
                        FLAG_MAP.remove(sessionId);
                    }
                })
                //流出现异常的时候执行
                .doOnError(new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) {
                        log.info("doOnError开始,异常:{}", sessionId);
                        FLAG_MAP.remove(sessionId);
                    }
                })
                //简单来说:来一个元素判断有没有sessionId,有的话继续输出,没有的话就停止输出
                /*
                takeWhile操作符的实际作用是：
                    逐个检查元素：对流中的每个元素进行条件测试
                    条件判断：只要条件为true，就继续传递元素；一旦条件为false，就停止传递后续所有元素
                    短路操作：一旦遇到第一个不满足条件的元素，就会中断整个流
                * */
                .takeWhile(new Predicate<ChatResponse>() {
                    @Override
                    public boolean test(ChatResponse chatResponse) {
                        //这个方法就是 拿sessionid去map里判断,有就返回value,没有就返回false
                        Boolean result = FLAG_MAP.getOrDefault(sessionId, false);
                        log.info("takeWhile开始执行,完成现在返回的是:{}", result);
                        return result;
                    }
                })
                //转换成ChatEventVO对象
                .map(respones ->{
                    return ChatEventVO.builder()
                            .eventData(respones.getResult().getOutput().getText())
                            .eventType(ChatEventTypeEnum.DATA.getValue())
                            .build();
                })
                .concatWith(Flux.just(
                        ChatEventVO.builder()
                                .eventType(ChatEventTypeEnum.STOP.getValue())
                                .build()
                ));
    }

    /**
     * 停止生成
     *思路:
     * 有恁多会话,你咋知道是停哪一个呢??
     * 所以就需要根据会话id(sessionId)来判断
     * 那怎么让流停止呢??
     * 跟多线程停止一个道理
     * 本来它是一直在执行是死循环
     * 现在给他个标签 一旦标签改变就停止 擦除掉
     * 所以需要个map去存这些
     * @param sessionId 会话id
     */
    @Override
    public void stop(String sessionId) {
        log.info("执行stop");
        FLAG_MAP.remove(sessionId);
    }
}